package bch.elasticsearch.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.*;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalClientExample {

    public static void main(String[] args) {
        // 连接到 Canal Server（注意这里的 address 和 port）
        CanalConnector connector = CanalConnectors.newSingleConnector(
            new InetSocketAddress("192.168.1.46", 11111), // canal server 地址和端口
            "example", // instanceName（destination）
            "", // 用户名（如果启用了）
            ""  // 密码（如果启用了）
        );

        try {
            connector.connect();
            connector.subscribe("test\\.user"); // 监听所有库所有表（你也可以写特定的，如 mydb\\.user）
            connector.rollback(); // 回滚到未确认的状态，重新获取数据

            while (true) {
                Message message = connector.getWithoutAck(100); // 批量拉取最多100条
                long batchId = message.getId();

                if (batchId != -1 && message.getEntries().size() > 0) {
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 确认消息处理完成
            }
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<Entry> entryList) {
        for (Entry entry : entryList) {
            if (entry.getEntryType() == EntryType.ROWDATA) {
                RowChange rowChange;
                try {
                    rowChange = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR parsing entry", e);
                }

                EventType eventType = rowChange.getEventType();
                System.out.println("=== " + eventType + " ===");

                for (RowData rowData : rowChange.getRowDatasList()) {
                    if (eventType == EventType.DELETE) {
                        printColumns(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT) {
                        printColumns(rowData.getAfterColumnsList());
                    } else {
                        System.out.println("before:");
                        printColumns(rowData.getBeforeColumnsList());
                        System.out.println("after:");
                        printColumns(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    }

    private static void printColumns(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " = " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}
